package com.zhao.demo.unbound.demo04_transformation.sample05_connect

import com.zhao.demo.Raytek
import org.apache.flink.streaming.api.scala.{ConnectedStreams, DataStream, SplitStream, StreamExecutionEnvironment}

/**
 * Description:使用DataStream的算子之connect将两个类型不同的流合并在一起,分别进行单独处理 <br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/12/8 17:52
 *
 * @author 柒柒
 * @version : 1.0
 */

object ConnectionStreamDemo {
  def main(args: Array[String]): Unit = {
    //1.环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //②实时读取流数据，计算，并显示结果
    import org.apache.flink.api.scala._


    val splitStream: SplitStream[Raytek] = env.socketTextStream("node01", 7775)
      .map(perInfo => {
        val arr = perInfo.split(",")
        val id = arr(0).trim
        val temperature = arr(1).trim.toDouble
        val name = arr(2).trim
        val timestamp = arr(3).trim.toLong
        val location = arr(4).trim
        Raytek(id, temperature, name, timestamp, location)
      }).split((rayteck: Raytek) => if (rayteck.temperature >= 36.3 && rayteck.temperature <= 37.2) Seq("正常") else Seq("异常"))

    //针对不同特征的旅客,进行不同的处理
    //a)从流中取出所有体温正常的旅客信息,进行处理
    val nomalStream: DataStream[(String, String)] = splitStream.select("正常")
      .map(perEle => (perEle.id, s"名字为<${perEle.name}>的旅客体温正常!!!!"))

    //b)从流中取出所有体温异常的旅客信息,进行处理
    val exceptionStream: DataStream[(String, String, String)] = splitStream.select("异常")
      .map(perEle => (perEle.id, perEle.name, s"该旅客体温异常,需要隔离!!!!"))

    //将上述两种类型的流合起来,进行集中式的分析处理,然后输出
    val connectedStream: ConnectedStreams[(String, String), (String, String, String)] = nomalStream.connect(exceptionStream)

    connectedStream.map(
      nomal => ("红外测温仪的id" + nomal._1, "旅客的信息是->" + nomal._2),
      exception => ("红外测温仪的id->" +exception._1, "旅客的名字是->" +exception._2, "报警信息是->" + exception._3)
    ).print()

    //3.启动
    env.execute()
  }
}
